package com.whiteseason.spark.GraphX

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD

object StructureOperate {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    conf.set("spark.testing.memory", "2147480000")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //设置顶点数据
    val vertexArray = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )
    val vertex: RDD[(VertexId, (String, Int))] = sc.parallelize(vertexArray)

    //设置边
    val edgeArray = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )
    val relationships: RDD[Edge[Int]] = sc.parallelize(edgeArray)

    //创建图1
    val graph = Graph(vertex, relationships)

    //创建图2
    val vertexArray2 = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50)),
      (7L, ("nuke",45))
    )
    val vertex2: RDD[(VertexId, (String, Int))] = sc.parallelize(vertexArray2)
    val edgeArray2 = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3),
      Edge(5L, 7L, 4),
      Edge(7L, 1L, 3),
      Edge(7L, 1L, 5),
      Edge(5L, 7L, 7)
    )
    val relationships2: RDD[Edge[Int]] = sc.parallelize(edgeArray2)

    val graph2 = Graph(vertex2, relationships2)

    //调用图1的mask方法，传入图2作为参数，观察返回的图3的信息
    println("---------------------返回的图3的信息------------------------")
    val graph3 = graph.mask(graph2)
    graph3.vertices.foreach(println(_))

    //基于上面的数据进行修改练习subgraph 、groupEdges 、reverses方法
    println("---------------------reverses反转图一信息------------------------")
    graph.reverse.edges.foreach(println(_))
    println("---------------------subgraph过滤图一小于30岁的顶点------------------------")
    val graph4 = graph.subgraph(vpred = (id, attr) => attr._2 > 30)
    graph4.vertices.collect.foreach(println(_))
    println("---------------------groupEdges合并图二相同边------------------------")
    println("旧边：")
    graph2.edges.collect.foreach(println(_))
    val ngraph2 = graph2.groupEdges(merge = (ed1, ed2) => (ed1 + ed2))
    println("新边：")
    ngraph2.edges.collect.foreach(println(_))
  }
}
